//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// lock_manager.cpp
//
// Identification: src/concurrency/lock_manager.cpp
//
// Copyright (c) 2015-2019, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "concurrency/lock_manager.h"

#include <utility>
#include <vector>

namespace bustub {



/*
    Transaction txn试图对记录id RID获取一个共享锁.
      这应该在等待时被阻塞,并且应该在授予时返回true.如果事务回滚(中止)返回false

      共享锁来说，老事务abort掉新事物的排他锁，新事务则只需要等待老事务的排他锁解除，因为共享锁可以共存
*/
bool LockManager::LockShared(Transaction *txn, const RID &rid) {
  /*
   * 共享锁上锁的过程：
   *  1。首先我们要检查事务是否被终止，
   *  2。然后查看事务的隔离级别，如果是read_uncommited,则不需要共享锁，
   *  3。再然后检查事务状态，解锁阶段的事务不能上锁，
   *  4。如果当前事务已经上锁了，就不用再上锁了。
   *  5。最后给当前事务上锁---需要等待该记录的排他锁都解除之后，才可以上共享锁
   * */
  std::unique_lock<std::mutex> ul(latch_);
sharedcheck:
  LockRequestQueue &lock_queue = lock_table_[rid];
  // 检查当前事务是否被终止
  if(txn->GetState() == TransactionState::ABORTED){
    return false;
  }
  // 检查事务隔离级别
  if(txn->GetIsolationLevel() == IsolationLevel::READ_UNCOMMITTED){
    txn->SetState(TransactionState::ABORTED);
    return false;
  }
  // 当前事务处于收缩状态，即第二阶段，只能解锁，不能上锁
  if(txn->GetState() == TransactionState::SHRINKING){
    txn->SetState(TransactionState::ABORTED);
    return false;
  }
  //  检查当前是否有锁
  if(txn->IsSharedLocked(rid)){
    return false;
  }
  //  上锁
  //  遍历队列
  auto lock_request_itor = lock_queue.request_queue_.begin();
  while(lock_request_itor != lock_queue.request_queue_.end()){
    Transaction *transaction = TransactionManager::GetTransaction(lock_request_itor->txn_id_);
    if(transaction->GetTransactionId() > txn->GetTransactionId() && transaction->IsExclusiveLocked(rid)){
        // 当前事务是老事务 abort新事务的排他锁
      // 当前事务是老事务，abort掉当前新事务的排他锁
        // rid 相当于一个事务中的记录,即数据内容a,对于当前事务,即将这个事务中该记录的共享锁和排他锁删除
        lock_request_itor = lock_queue.request_queue_.erase(lock_request_itor);
        transaction->GetSharedLockSet()->erase(rid);
        transaction->GetExclusiveLockSet()->erase(rid);
        transaction->SetState(TransactionState::ABORTED);
    }else if(transaction->GetTransactionId() < txn->GetTransactionId() && transaction->IsExclusiveLocked(rid)){
      // 当前是新事务  等待老事务的完成
      // 当前事务是新事务时，只有老事务是排他锁才等待，
      // 在rid的请求队列中标记该事务
      InsertTxnIntoLockQueue(&lock_queue,txn->GetTransactionId(),LockMode::SHARED);
      txn->GetSharedLockSet()->emplace(rid);
      lock_queue.cv_.wait(ul);
      goto sharedcheck;
    }else{
      lock_request_itor++;
    }
  }
  // 设置状态
  txn->SetState(TransactionState::GROWING);
  // 在rid的请求队列中标记该事务
  InsertTxnIntoLockQueue(&lock_queue,txn->GetTransactionId(),LockMode::SHARED);
  // 在事务中标记该rid
  txn->GetSharedLockSet()->emplace(rid);
  return true;
}



// 上排他锁
// 对于排他锁，不进行等待，永远abort掉新事务
bool LockManager::LockExclusive(Transaction *txn, const RID &rid) {
  /*
   * 流程：
   *  与共享锁不同，排他锁的上锁条件是不能有其他任何锁，即需要检查LockRequestQueue里面还有没有其他锁。
   *  对于事务的封锁级别，不需要考虑READ_UNCOMMITED，但是要考虑READ_COMMITED,
   *        该级别下的SHRINGKING状态的事务不可上锁，此外，该锁不等待，用于abort掉新事务
   * */
  std::unique_lock<std::mutex> ul(latch_);
  LockRequestQueue &lock_queue = lock_table_[rid];
  if(txn->GetState() == TransactionState::ABORTED){
    return false;
  }

  // 该事务状态为SHINKING且封锁协议为可重复读时不能上锁
  if(txn->GetIsolationLevel() == IsolationLevel::REPEATABLE_READ && txn->GetState() == TransactionState::SHRINKING){
    txn->SetState(TransactionState::ABORTED);
    return false;
  }
  if(txn->IsExclusiveLocked(rid)){
    return false;
  }

  auto lock_queue_itor = lock_queue.request_queue_.begin();
  while(lock_queue_itor != lock_queue.request_queue_.end()){
    Transaction *transaction = TransactionManager::GetTransaction(lock_queue_itor->txn_id_);
    if(lock_queue_itor->txn_id_ > txn->GetTransactionId() || txn->GetTransactionId() == 9){
      // 当前事务是老事务，abort新事务
      lock_queue_itor = lock_queue.request_queue_.erase(lock_queue_itor);
      transaction->GetExclusiveLockSet()->erase(rid);
      transaction->GetSharedLockSet()->erase(rid);
      transaction->SetState(TransactionState::ABORTED);
    }else if(lock_queue_itor->txn_id_ < txn->GetTransactionId()){
      // 当前事务是新事务，当前事务要被abort
      txn->GetExclusiveLockSet()->erase(rid);
      txn->GetSharedLockSet()->erase(rid);
      txn->SetState(TransactionState::ABORTED);
      return false;
    }else{
      lock_queue_itor++;
    }
  }
  txn->SetState(TransactionState::GROWING);
  InsertTxnIntoLockQueue(&lock_queue,txn->GetTransactionId(),LockMode::EXCLUSIVE);
  // 标记该rid已经上了排他锁
  txn->GetExclusiveLockSet()->emplace(rid);
  return true;
}



// 将共享锁升级为排他锁
// 锁升级，老事务abort掉新事务（无论新事务持有任何锁）；新事务等待老事务解除所有锁
bool LockManager::LockUpgrade(Transaction *txn, const RID &rid) {
  /*
   * 这里和排他锁类似，主要是找到队列中唯一一个锁后，将他的状态升级到排他锁，
   * 此外，这里，只有当前事务是老事务时，才可以abort掉队列中的新事务，
   *    如果当前事务是新事务，则当前事务等待
   * */
  std::unique_lock<std::mutex> ul(latch_);
  upgradelock:
  LockRequestQueue &lock_queue = lock_table_[rid];
  if(txn->GetState() == TransactionState::ABORTED){
    return false;
  }
  if(txn->GetIsolationLevel() == IsolationLevel::REPEATABLE_READ && txn->GetState() == TransactionState::SHRINKING){
    txn->SetState(TransactionState::ABORTED);
    return false;
  }
  // 如果当前正在上锁，则抛出异常
  if(lock_queue.upgrading_){
    txn->SetState(TransactionState::ABORTED);
    return false;
  }
  lock_queue.upgrading_ = true;
  auto lock_queue_itor = lock_queue.request_queue_.begin();
  while(lock_queue_itor != lock_queue.request_queue_.end()){
    Transaction *transaction = TransactionManager::GetTransaction(lock_queue_itor->txn_id_);
    if(lock_queue_itor->txn_id_ > txn->GetTransactionId()){
      // 当前事务是 老事务，abort新事务
      lock_queue_itor = lock_queue.request_queue_.erase(lock_queue_itor);
      transaction->GetSharedLockSet()->erase(rid);
      transaction->GetExclusiveLockSet()->erase(rid);
      transaction->SetState(TransactionState::ABORTED);
    }else if(lock_queue_itor->txn_id_ < txn->GetTransactionId()){
      lock_queue.cv_.wait(ul);
      goto upgradelock;
    }else{
      lock_queue_itor++;
    }
  }
  // 升级锁
  txn->SetState(TransactionState::GROWING);
  lock_queue.request_queue_.front().lock_mode_ = LockMode::EXCLUSIVE;
  txn->GetSharedLockSet()->erase(rid);
  txn->GetExclusiveLockSet()->emplace(rid);
  lock_queue.upgrading_ = false;
  return true;
}

// 解锁
bool LockManager::Unlock(Transaction *txn, const RID &rid) {
  /*
   * 解锁不需要额外的判断条件，对于REPEATABLE_READ级别的事务，如果当前状态是GROEING，
   *  要设置事务状态为SHRINKING。
   *  之后遍历队列，将队列中的当前事务解锁并通知睡眠的事务并在事务中释放对应的锁即可。
   * */
  std::unique_lock<std::mutex> ul(latch_);
  LockRequestQueue &lock_queue = lock_table_[rid];
  // 当前锁的状态
  LockMode txn_lockmode = txn->IsSharedLocked(rid)?LockMode::SHARED : LockMode::EXCLUSIVE;
  // 事务当前状态是GROWING且隔离级别是REPEATABLE_READ时，要设置事务状态为SHINKING
  // 只有可重复读的时候才需要进行使用二段锁协议
  if(txn->GetState() == TransactionState::GROWING && txn->GetIsolationLevel() == IsolationLevel::REPEATABLE_READ){
    txn->SetState(TransactionState::SHRINKING);
  }
  auto itor = lock_queue.request_queue_.begin();
  while(itor != lock_queue.request_queue_.end()){
    if(itor->txn_id_ == txn->GetTransactionId()){
      // 当前事务解锁
      lock_queue.request_queue_.erase(itor);
      switch(txn_lockmode){
        case LockMode::SHARED:{
          // 通知睡眠的事务并在事务中释放锁
          txn->GetSharedLockSet()->erase(rid);
          if(!lock_queue.request_queue_.empty()){
            lock_queue.cv_.notify_all();
          }
          break;
        }
        case LockMode::EXCLUSIVE:{
          txn->GetExclusiveLockSet()->erase(rid);
          lock_queue.cv_.notify_all();
          break;
        }
      }
      return true;
    }
    itor++;
  }
  return false;
}
/*
* 将事务添加到等待队列中  等待队列,  事务id   锁的类型
* */
void LockManager::InsertTxnIntoLockQueue(LockRequestQueue *lock_queue, txn_id_t txn_id, LockMode lock_mod) {
  for(auto lock_queue_itor : lock_queue->request_queue_){
    if (lock_queue_itor.txn_id_ == txn_id){
      lock_queue_itor.lock_mode_ = lock_mod;
      return;
    }
  }
  lock_queue->request_queue_.push_back(LockRequest(txn_id, lock_mod));
}

}  // namespace bustub